package com.gallop.common.reliable;

import com.gallop.common.reliable.config.ReliableDeliveryConfig;
import com.gallop.common.lock.DistributeLocker;
import lombok.Data;
import lombok.extern.slf4j.Slf4j;
import org.redisson.api.RBucket;
import org.redisson.api.RedissonClient;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import org.springframework.util.StringUtils;

import java.util.Collection;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;

@Component
@Slf4j
public class RedissonReliableDeliveryHandler implements ReliableDeliveryStrategy {

    @Autowired
    private ReliableDeliveryConfig reliableDeliveryConfig;
    @Autowired
    private RedissonClient redissonClient;

    @Override
    public boolean handleBeforeSend(String topic, String id, Object message, Collection<String> listeners) {
        AtomicBoolean result = new AtomicBoolean(true);
        listeners.parallelStream()
                .map(group -> StringUtils.hasLength(StringUtils.trimAllWhitespace(group)) ? group : reliableDeliveryConfig.getDefaultGroup())
                .forEach(group -> {
                    try {
                        String messageKey = appendMessageKey(topic, id, group);
                        DistributeLocker.getInstance().lock(DistributeLocker.DistributedLockOption.builder(appendKey(messageKey, KeyAppender.RELIABLE_MUTEX))
                                .setMaxWait(TimeUnit.SECONDS.toMillis(reliableDeliveryConfig.getLockTimeout() * 2L))
                                .setExpireAfter(TimeUnit.SECONDS.toMillis(reliableDeliveryConfig.getLockTimeout()))
                                .setMaxWaitTimeUint(TimeUnit.MILLISECONDS)
                                .setExpireTimeUnit(TimeUnit.MILLISECONDS)
                                .setOnSuccess(() -> {
                                    RBucket<ReliablePayload> mutexBucket = redissonClient.getBucket(appendKey(messageKey, KeyAppender.RELIABLE_MSG));
                                    if (mutexBucket.get() != null) {
                                        // 重复发送
                                        return;
                                    }
                                    // 记录接收方
                                    ReliablePayload payload = new ReliablePayload();
                                    payload.setAck(false);
                                    payload.setGroup(group);
                                    payload.setId(id);
                                    payload.setTopic(topic);
                                    payload.setMessage(message);
                                    mutexBucket.set(payload, reliableDeliveryConfig.getMutexKeepSeconds(), TimeUnit.SECONDS);
                                })
                                .setOnError(error -> {
                                    log.error("mq reliable process error.", error);
                                    result.set(true);
                                })
                                .setOnFail(key -> {
                                    log.error("mq reliable fail to lock. key={}", key);
                                    result.set(true);
                                })
                                .build());
                    } catch (Exception e) {
                        log.error("", e);
                        result.set(false);
                    }
                });
        return result.get();
    }


    @Override
    public boolean handleBeforeAck(String topic, String group, String id, Object message) {
        AtomicBoolean result = new AtomicBoolean(true);
        try {
            String messageKey = appendMessageKey(topic, id, group);
            DistributeLocker.getInstance().lock(DistributeLocker.DistributedLockOption.builder(appendKey(messageKey, KeyAppender.RELIABLE_MUTEX))
                    .setMaxWait(TimeUnit.SECONDS.toMillis(reliableDeliveryConfig.getLockTimeout() * 2L))
                    .setExpireAfter(TimeUnit.SECONDS.toMillis(reliableDeliveryConfig.getLockTimeout()))
                    .setMaxWaitTimeUint(TimeUnit.MILLISECONDS)
                    .setExpireTimeUnit(TimeUnit.MILLISECONDS)
                    .setOnSuccess(() -> {
                        RBucket<ReliablePayload> messageBucket = redissonClient.getBucket(appendKey(messageKey, KeyAppender.RELIABLE_MSG));
                        ReliablePayload payload = messageBucket.get();
                        if (payload != null && payload.isAck()) {
                            log.error("ReliableDeliveryStrategy duplicated mq message will be blocked. msg={},topic={},group={}", message, topic, group);
                            result.set(false);
                            return;
                        }
                        if (payload == null) {
                            payload = new ReliablePayload();
                            payload.setGroup(group);
                            payload.setId(id);
                            payload.setTopic(topic);
                            payload.setMessage(message);
                        }
                        // 记录接收方
                        payload.setAck(true);
                        messageBucket.set(payload, this.reliableDeliveryConfig.getMutexKeepSeconds(), TimeUnit.SECONDS);
                    })
                    .setOnError(error -> {
                        log.error("mq reliable ack process error.", error);
                        result.set(false);
                    })
                    .setOnFail(key -> {
                        log.error("mq reliable ack fail to lock. key={}", key);
                        result.set(false);
                    })
                    .build());
        } catch (Exception e) {
            log.error("", e);
            result.set(false);
        }
        return result.get();
    }

    private String appendMessageKey(String topic, String id, String group) {
        return KeyAppender.RELIABLE_PREFIX.toKey() + topic + ":" + group + ":" + id;
    }

    private String appendKey(String key, KeyAppender suffix) {
        return key + ":" + suffix.val;
    }

    @Data
    public static class ReliablePayload {

        private String topic;
        private String id;
        private String group;
        private boolean ack;
        private Object message;
    }

    public enum KeyAppender {
        RELIABLE_PREFIX("$$rd"),
        RELIABLE_MSG("msg"),
        RELIABLE_MUTEX("mutex"),
        ;

        private final String val;

        KeyAppender(String val) {
            this.val = val;
        }

        public String toKey() {
            return this.val + ":";
        }
    }
}
